/*
 * This part is the most important part of the Frame.
 *
 * This file contain the IO engine and do most of the
 * stuff work for the module.Doing these stuff work automaticly
 * and transparently for the module can enhance the safety and
 * compability.That means the module writer just need to focus on 
 * the functions of his module.
 *
 */

/*
 * ChangeLog:
 * 2012-3-12: 添加了物件池。
 *
 */

/*
 * 这个文件(指的是burst_react.c)在bus服务器项目里的地位十分重要。
 * 
 * 这里的代码是服务器最核心的引擎，需要高的负载控制，
 * 否则会造成对客户端的响应慢，
 * 吞吐量低下，甚至导致系统交互性反应迟钝，所以合理的设计是最重要的。
 *
 * 在C10K问题里(原文:http://www.kegel.com/c10k.html
 *                汉化:http://wenku.baidu.com/view/33e41da6f524ccbff12184f9.html)，
 * 就曾经说过1万并发是怎么实现的，
 * 在这里我就先说说总概括有多少种模型。
 *
 * 1. 每个连接都分配一个线程:
 *    这就是说，一个主线程负责accept连接，然后为每个得到的套接字分配1条线程。
 *    很简单。
 *  
 *    好处：写代码很方便。
 *    坏处：会有上下文切换的负载，多连接的时候电脑交互性还会下降。
 *
 * 2. 事件驱动的非阻塞IO：
 *    其实这是一个状态机，线程创建数最好根据cpu个数决定。
 *
 *    好处：这么多方法里有最高并发。
 *    坏处：代码书写比其他方法困难。
 *
 * 在这里，Bus服务器使用的是事件驱动的非阻塞IO, 这是为了获得最好的性能。
 * 思路：
 *    首先，一个最基本的状态机形式，包括6种状态，如下。
        #define BUS_IO_READING 1
        #define BUS_IO_SLEEPING 2
        #define BUS_IO_WRITING 4
        #define BUS_IO_ACCEPTING 8
        #define BUS_IO_CONNECTING 16
        #define BUS_IO_ERROR 32
 *    状态机根据不同的状态，做出不同的反应。
 *
 *    对于解决io瓶颈，我们使用了一个很特别的方法。
 *    首先我们为每一个bus_event_t初始化池对象，默认是4个对象能获得较好的效果。
 *    对每次模块投递上来的io请求，我们会用队列的方式处理，并且为每一次请求
 *    从对象池分配一个io对象，io完成的时候，io对象会被放回空闲链表。
 *    另外，我们有一个策略，如果当前对象池没有空闲物件，那么对应的bus_event_t
 *    将会进入瞌睡状态，直到物件池资源不紧缺的时候会自动解除。瞌睡状态下：
 *    套接字的io操作就会变成半双工模式。
 *    
 *    IO对象池可以增长或收缩，但这些操作必须手动进行，服务器内核不会帮模块
 *    做增长或收缩。
 *
 * -------------------Coding and remarking By : QQ:3766691--------------------------- *
 */

/*
 * 对bus_event_t 结构体进行的解说：
 * 结构体原型 :

    struct bus_event {
        u_int sockfd;
        void *data;
        u_int type;
        u_int event;
        u_int status;
        union {
            bus_pipe_t pipe;
        };
        bus_io_poll_t io_object;
        bus_io_sender_t sender;
        bus_handle_event_t *handler;
        struct sockaddr_in sa;
        struct bus_event *father;
        struct list_head list;
    };

     * @sockfd - 对应bus_event结构体的套接字

     * @data - 对应bus_event结构体的附带数据（供插件使用）

     * @type - 对应bus_event结构体的用途（一般不会怎么有影响）

     * @event - 对应bus_event结构体的套接字发生过的io事件

     * @status - 对应bus_event结构体的套接字进行中/开放的io操作

     * @pipe - 对应bus_event结构体的套接字的管道（服务器全自动专用io托管）。

     * @io_object - 对应bus_event结构体的套接字的IO对象池。

     * @sender - 对应bus_event结构体的套接字的IO事件状态信息（旧）。

     * @handler - 对应bus_event结构体的套接字的IO完成后调用的call back。

     * @sa - 对应bus_event结构体的套接字的目标连接的信息。

     * @father - bus_event结构体问：“我的父亲呢？”， 答1：“你没有父亲”， 答2：“你的父亲在这。”。

     * @list - 发生IO事件的bus_event结构体的队列。
 *
 */

#include <burst_kernel.h>


/*
 * bus_check_connect -Check the connecting status of bus_event_t structure.
 * @ev:        bus_event_t structure
 *
 * We check it by using getpeername because we can only get the remote
 * struct sockaddr structure when the connection was established.If there 
 * is no any connection or no client, getpeername will return -1.
 *
 * Return NULL means the connection was established, otherwise return -1.
 *
 */
static int bus_check_connect(bus_event_t *ev)
{

    int addr_len = sizeof(struct sockaddr);
    struct sockaddr_in sa = {0};

    if(getpeername(ev->sockfd, (struct sockaddr *)&sa, (int *)&addr_len) == -1) {
        return -1;
    }
    
    return 0;

}


/* --- IO Object pool code. --- */

/*
 * bus_event_io_poll_dtor - The destructor of the object poll.
 * @ev:        bus_event_t structure
 *
 * Iterate the 3 linked-list and free all the object.
 *
 */
static void bus_event_io_poll_dtor(bus_event_t *ev)
{
    bus_io_object_t *tmp;
    struct list_head *cur;

    /*
     * 遍历idle链表并释放。
     * 
     * 这个过程主要是用于析构。
     */
    list_for_each(cur, &ev->io_object.idle) {

        tmp = (bus_io_object_t *)container_of(cur, bus_io_object_t, node);

        list_del(cur);

        dlfree(tmp);

    }

    /*
     * 遍历using_r链表并释放。
     * 
     * 这个过程主要是用于析构。
     */
    list_for_each(cur, &ev->io_object.using_r) {

        tmp = (bus_io_object_t *)container_of(cur, bus_io_object_t, node);

        list_del(cur);

        dlfree(tmp);

    }

    /*
     * 遍历using_w链表并释放。
     * 
     * 这个过程主要是用于析构。
     */
    list_for_each(cur, &ev->io_object.using_w) {

        tmp = (bus_io_object_t *)container_of(cur, bus_io_object_t, node);

        list_del(cur);

        dlfree(tmp);

    }
    memset(&ev->io_object, NULL, sizeof(bus_io_poll_t)); 
}

/*
 * bus_event_io_poll_ctor - The constructor of the object poll.
 * @ev:        bus_event_t structure
 * @object: The nr of object to be initialized.
 *
 * Initializing and adding the nr of object to the poll.
 *
 * Return NULL means succeeded, otherwise failed for -1.
 *
 */
static int bus_event_io_poll_ctor(bus_event_t *ev, int object)
{
    int ret = -1;
    int i;
    bus_io_object_t *tmp;

    if(ev->io_object.obj_count)
        goto out;

    INIT_LIST_HEAD(&ev->io_object.idle);

    INIT_LIST_HEAD(&ev->io_object.using_r);
    INIT_LIST_HEAD(&ev->io_object.using_w);

    for(i = 0;i < object;i++, ev->io_object.obj_count++) {

        tmp = dlzalloc(sizeof(bus_io_object_t));
        if(!tmp) {

            /*
             * Memory allocation error! Destruct it.
             *
             */
            bus_event_io_poll_dtor(ev);

            goto out;

        }

        list_add(&tmp->node, &ev->io_object.idle);

    }
    ret = 0;
out:
    return ret;
}

/*
 * bus_event_io_poll_ctor - The constructor of the object poll.
 * @ev:        bus_event_t structure
 * @object: The nr of object to be grown.
 *
 * Adding the nr of object to the poll.
 *
 * Return NULL always.
 *
 */
static int bus_event_io_poll_grow(bus_event_t *ev, int object)
{

    int i;
    bus_io_object_t *tmp;

    for(i = 0;i < object;i++, ev->io_object.obj_count++) {

        tmp = dlzalloc(sizeof(bus_io_object_t));
        if(!tmp) {

            goto out;

        }

        list_add(&tmp->node, &ev->io_object.idle);

    }

out:
    return 0;

}

/*
 * bus_event_io_poll_shrink - 为对象池进行收缩。.
 * @ev:        bus_event_t structure
 * @object: The nr of object to be shrink.
 *
 * Removing the nr of object from the poll.
 *
 * Return NULL always.
 *
 */
static int bus_event_io_poll_shrink(bus_event_t *ev, int object)
{
    int i = 0;
    bus_io_object_t *tmp;
    struct list_head *cur;
    list_for_each(cur, &ev->io_object.idle) {
        if(i == object)
            goto out;
        tmp = (bus_io_object_t *)container_of(cur, bus_io_object_t, node);
        list_del(cur);
        dlfree(tmp);
        i++;
        ev->io_object.obj_count--;
    }
out:
    return 0;
}

/*
 * bus_event_io_poll_get_object - 从对象池分配对象。
 * @ev:        bus_event_t structure
 *
 * 从对象池分配1个对象.
 *
 * Return the object we get.
 *
 */
static bus_io_object_t *bus_event_io_poll_get_object(bus_event_t *ev)
{
    int i = 0;
    bus_io_object_t *tmp = 0;
    struct list_head *cur;
    if(ev->io_object.obj_count == 0) {
        goto out;
    }
    cur = ev->io_object.idle.next;
    list_del(cur);
    tmp = (bus_io_object_t *)container_of(cur, bus_io_object_t, node);
    tmp->bufptr = tmp->buf_start;
    ev->io_object.obj_count--;
out:
    return tmp;
}

/*
 * bus_event_io_poll_put_object - 从对象池释放对象。
 * @ev:        bus_event_t structure
 *
 * 从对象池释放1个对象.
 *
 */
static void bus_event_io_poll_put_object(bus_event_t *ev, bus_io_object_t *object)
{
    list_add(&object->node, &ev->io_object.idle);
    ev->io_object.obj_count++;
}


static void bus_event_io_poll_insert_queue_first
            (bus_event_t *ev, struct list_head *head, bus_io_object_t *object)
{
    list_add(&object->node, head);
}

static void bus_event_io_poll_insert_queue_last
            (bus_event_t *ev, struct list_head *head, bus_io_object_t *object)
{
    list_add_tail(&object->node, head);
}

static bus_io_object_t *bus_event_io_poll_get_queue_first
                        (bus_event_t *ev, struct list_head *head)
{
    struct list_head *tmp = head->next;
    bus_io_object_t *ret = 0;
    if(list_empty(head)) {
        goto out;
    }
    ret = (bus_io_object_t *)container_of(tmp, bus_io_object_t, node);
out:
    return ret;
}

static bus_io_object_t *bus_event_io_poll_get_queue_last
                        (bus_event_t *ev, struct list_head *head)
{
    struct list_head *tmp = head->prev;
    bus_io_object_t *ret = 0;
    if(list_empty(head)) {
        goto out;
    }
    ret = (bus_io_object_t *)container_of(tmp, bus_io_object_t, node);
out:
    return ret;
}

static void bus_event_io_poll_remove_queue_first
            (bus_event_t *ev, struct list_head *head)
{
    struct list_head *tmp = head->next;

    if(list_empty(head)) {
        goto out;
    }

    list_del(tmp);

out:
    return;
}

static void bus_event_io_poll_remove_queue_last
            (bus_event_t *ev, struct list_head *head)
{
    struct list_head *tmp = head->prev;
    if(list_empty(head)) {
        goto out;
    }
    list_del(tmp);
out:
    return;
}

static int bus_event_io_poll_remain_object(bus_event_t *ev)
{
    return ev->io_object.obj_count;
}

/* End --- IO Object pool code. --- */


/* --- IO status kernel call for module. --- */

int bus_event_get_event(bus_event_t *ev)
{
    return ev->io_object.obj_using->event;
}

/*
 * bus_event_get_buf - Get io buffer pointer from the bus_event_t structure.
 * @ev:        bus_event_t structure
 *
 * Return the ev->sender.buf_start;
 */
uint8_t *bus_event_get_buf(bus_event_t *ev)
{
    return ev->io_object.obj_using->buf_start;
}

/*
 * bus_event_get_size - Get io buffer size from the bus_event_t structure.
 * @ev:        bus_event_t structure
 *
 * Return the ev->sender.remain;
 */
int bus_event_get_size(bus_event_t *ev)
{
    return ev->io_object.obj_using->len;
}
/*
 * bus_event_get_data - Get data from the bus_event_t structure.
 * @ev:        bus_event_t structure
 *
 * Return the ev->data;
 */
void *bus_event_get_data(bus_event_t *ev)
{
    return ev->data;
}

/*
 * bus_event_set_data - Set data from the bus_event_t structure.
 * @ev:        bus_event_t structure
 * @data:        The data pointer
 *
 */
void bus_event_set_data(bus_event_t *ev, void *data)
{
    ev->data = data;
}

/* End --- IO status kernel call for module. --- */

/*
 * bus_callback - Call the callback function.
 * @ev:        bus_event_t structure
 */
void bus_callback(bus_event_t *ev)
{
    
    if(!ev->handler)
        return;
    ev->handler(ev);

}

/*
 * bus_get_callback - Get callback function pointer.
 * @ev:        bus_event_t structure
 */
bus_handle_event_t *bus_get_callback(bus_event_t *ev)
{
    
    return ev->handler;

}

/*
 * bus_set_callback - Set callback function pointer.
 * @ev:            bus_event_t structure
 * @handler:    The callback function pointer.
 */
void bus_set_callback(bus_event_t *ev, bus_handle_event_t *handler)
{
    
    ev->handler = handler;

}

/*
 * bus_get_socket - Safely get a bus_event_t structure from Kernel.
 * @type:        Socket type
 * @proto:        Socket protocol
 *
 * If succeed, return a bus_event_t structure, otherwise return NULL.
 */
bus_event_t *bus_get_socket(int type, int proto)
{

    bus_event_t *ev;
    int s = socket(AF_INET, type, proto);
    if(s < 0) {
        return NULL;
    }

    ev = (bus_event_t *)dlzalloc(sizeof(bus_event_t));
    if(!ev) {
        return NULL;
    }

    ev->sockfd = s;

    return ev;

}

/*
 * bus_close_socket - Safely close a bus_event_t structure from Kernel.
 * @ev:        The pointer of the bus_event_t structure.
 *
 */
void bus_close_socket(bus_event_t *ev)
{
    /*
     * Free the object pool of ev.
     */
    bus_event_io_poll_dtor(ev);

    /*
     * We have to delete the sockfd of ev from the event module.
     */
    bus_event->del(0, ev);

    closesocket(ev->sockfd);

    /*
     * Free the bus_event_t structure.
     */
    dlfree(ev);

    return;

}

/*
 * bus_add_binding - Bind a bus_event_t structure from Kernel with a struct sockaddr_in structure.
 * @ev:        The pointer of the bus_event_t structure.
 * @sa:        The pointer of the struct sockaddr_in structure.
 *
 * If succeed, return NULL, otherwise return -1.
 */
int bus_add_binding(bus_event_t *ev, struct sockaddr_in *sa)
{

    return bus_io->bind(ev->sockfd, (struct sockaddr *)sa, sizeof(struct sockaddr));

}

/*
 * bus_add_sending - Add a sending io request on the bus_event_t structure from Kernel.
 * @ev:        The pointer of the bus_event_t structure.
 * @buf:    The Buffer.
 * @len:    The length of data.
 * @rsv:    Unused.
 *
 */
void bus_add_sending(bus_event_t *ev, 
                char *buf, 
                int len, 
                int rsv)
{
    if(1) {
        bus_io_object_t *obj = bus_event_io_poll_get_object(ev);
        bus_event_io_open(ev, BUS_IO_WRITING);
        obj->event = BUS_IO_WRITING;

        /*
         * Copy the data to the io buffer.
         * 
         * XXX: It is quite a bad work i thought. ;) One time i should find a replacement
         *     to replace it.
         */
        memcpy(obj->bufptr, buf, len);
        obj->len = len;
        bus_event_io_poll_insert_queue_last(ev, &ev->io_object.using_w, obj);
    }
    return;
}

/*
 * bus_add_recving - Add a receiving io request on the bus_event_t structure from Kernel.
 * @ev:        The pointer of the bus_event_t structure.
 * @buf:    Unused.
 * @len:    Unused.
 * @rsv:    Unused.
 *
 */
void bus_add_recving(bus_event_t *ev, 
                char *buf, 
                int len, 
                int rsv)
{
    ev->io_object.rsv.event = BUS_IO_READING;
    bus_event_io_open(ev, BUS_IO_READING);
    return;
}

/*
 * bus_add_connect - Add a connecting io request on the bus_event_t structure from Kernel.
 * @ev:            The pointer of the bus_event_t structure.
 * @sa:            The pointer of the struct sockaddr_in structure.
 * @addr_len:    Unused.
 *
 */
void bus_add_connecting(bus_event_t *ev, 
                struct sockaddr *sa,
                int addr_len)
{
    bus_event->add(0, ev);

    bus_io->set_non_block(ev->sockfd, 1);

    bus_event_io_open(ev, BUS_IO_CONNECTING);

    bus_io->connect(ev->sockfd, sa, sizeof(struct sockaddr));

    return;

}

/*
 * bus_add_accept - Add a accepting io request on the bus_event_t structure from Kernel.
 * @ev:            The pointer of the bus_event_t structure.
 * @sa:            The pointer of the struct sockaddr_in structure.
 * @addr_len:    Unused.
 *
 */
void bus_add_accepting(bus_event_t *ev, 
                struct sockaddr *sa,
                int addr_len)
{
    bus_event->add(0, ev);

    bus_io->set_non_block(ev->sockfd, 1);

    bus_event_io_open(ev, BUS_IO_ACCEPTING);

    bus_io->listen(ev->sockfd, 10);

    return;

}



/*
 * bus_process_sending - Process a sending io request on the bus_event_t structure from Kernel.
 * @ev:            The pointer of the bus_event_t structure.
 *
 */
static int bus_process_sending(bus_event_t *ev)
{
    int err;
    int ret = 0;
    bus_io_object_t *obj = bus_event_io_poll_get_queue_first(ev, &ev->io_object.using_w);

    /*
     *    看看io事件是否可写
     */
    if(bus_event_bit(ev, BUS_IO_WRITING)) {

        err = bus_io->send(ev->sockfd, \
                        (char *)obj->bufptr, \
                        obj->len, \
                        0);

        if(err <= 0) {

            bus_event->del(0, ev);

            ret = -1;
            bus_event_set(ev, BUS_IO_ERROR);

            bus_callback(ev);
            goto out;

        }

        obj->bufptr += err;
        obj->len -= err;
        if(obj->len <= 0) {

            bus_event_io_poll_remove_queue_first(ev, &ev->io_object.using_w);

            ev->io_object.obj_using = obj;

            bus_event_io_close(ev, BUS_IO_WRITING);

            bus_callback(ev);

            bus_event_io_poll_put_object(ev, obj);

        }
        bus_event_unset(ev, BUS_IO_WRITING);
    }

out:
    if(ret < 0)
        bus_close_socket(ev);
    return ret;

}

/*
 * bus_process recving - Process a recving io request on the bus_event_t structure from Kernel.
 * @ev:            The pointer of the bus_event_t structure.
 *
 * Remark:    The received data is put into sender.buf_start.
 *            The size of data received is put into ev->sender.remain .
 */
static int bus_process_recving(bus_event_t *ev)
{
    int err;
    int ret = 0;
    bus_io_object_t *obj;

    /*
     *    看看是否有读请求。
     */
    if(bus_event_bit(ev, BUS_IO_READING)) {

        /* 
         * We do this to check whether there is writing operation or not.
         * We don't want to let the receive operation to replace the buf
         * when where is writing operation.
         * Just return NULL and report the caller as a succeeded IO operation.
         * In fact we wont call the callback.
         *
         * PS: I 'm using Gedit only :) The high-lighting color is not as sharp as Emacs.
         */
        if(list_empty(&ev->io_object.idle)) {
            goto out;
        }

        ev->io_object.obj_using = &ev->io_object.rsv;
        obj = ev->io_object.obj_using;

        obj->bufptr = obj->buf_start;

        err = bus_io->recv(ev->sockfd, \
                        (char *)obj->buf_start, \
                        BUS_BUF_SIZE, \
                        0);

        if(err <= 0) {

            bus_event->del(0, ev);

            ret = -1;
            bus_event_set(ev, BUS_IO_ERROR);

            bus_callback(ev);
            goto out;

        }

        obj->len = err;

        bus_event_io_close(ev, BUS_IO_READING);

        bus_callback(ev);

        bus_event_unset(ev, BUS_IO_READING);

    }

out:
    if(ret < 0)
        bus_close_socket(ev);

    return ret;

}

/*
 * bus_process_connecting - Process a connecting io request on the bus_event_t structure from Kernel.
 * @ev:            The pointer of the bus_event_t structure.
 *
 */
static int bus_process_connecting(bus_event_t *ev)
{
    int ret = 0;

    if(bus_event_bit(ev, BUS_IO_CONNECTING)) {

        int tmp_size = sizeof(struct sockaddr);

        bus_event_io_close(ev, BUS_IO_CONNECTING);

        if(bus_check_connect(ev) < 0) {

            bus_event_set(ev, BUS_IO_ERROR);

            bus_callback(ev);

            bus_event_unset(ev, BUS_IO_CONNECTING);
            bus_close_socket(ev);

        } else {

            bus_callback(ev);

            bus_event_unset(ev, BUS_IO_CONNECTING);

        }

    }

out:
    return ret;

}

/*
 * bus_process_accepting - Process a accepting io request on the bus_event_t structure from Kernel.
 * @ev:            The pointer of the bus_event_t structure.
 *
 */
static int bus_process_accepting(bus_event_t *ev)
{
    int ret = 0;

    if(bus_event_bit(ev, BUS_IO_ACCEPTING)) {

        bus_event_t *leaf_ev = (bus_event_t *)dlzalloc(sizeof(bus_event_t));

        int tmp_size = sizeof(struct sockaddr);

        int getfd = bus_io->accept(ev->sockfd, (struct sockaddr *)&leaf_ev->sa, &tmp_size);
        if(getfd == SOCKET_ERROR) {

            ret = SOCKET_ERROR;

            dlfree(leaf_ev);

            goto out;

        }

        leaf_ev->sockfd = getfd;
        leaf_ev->father = ev;

        bus_set_callback(leaf_ev, ev->handler);
        bus_callback(leaf_ev);

        bus_event_unset(ev, BUS_IO_ACCEPTING);
    }

out:
    return ret;

}


/* --- Sliding IO --- */

/*
 * Sliding IO means to call the module back automaticly and the
 * module *DO NOT* need to call bus_add_recving() anymore.
 *
 * This can help to decrease the size of code.
 *
 * In sliding IO, when a write request is done Kernel will not call you anymore.
 * unless some error happened.
 */

/*
 * bus_process_reading_sliding - 
 * Process a recving io request on the bus_event_t structure from Kernel(Sliding IO).
 *
 * @ev:            The pointer of the bus_event_t structure.
 *
 * Remark:    The received data is put into sender.buf_start.
 *            The size of data received is put into ev->sender.remain .
 */
static void bus_process_reading_sliding(bus_event_t *ev)
{
    int err;
    int ret = 0;
    bus_io_object_t *obj;

    /*
     *    看看是否有读请求。
     */
    if(bus_event_bit(ev, BUS_IO_READING)) {

        /* 
         * We do this to check whether there is writing operation or not.
         * We don't want to let the receive operation to replace the buf
         * when where is writing operation.
         * Just return NULL and report the caller as a succeeded IO operation.
         * In fact we wont call the callback.
         *
         * PS: I 'm using Gedit only :) The high-lighting color is not as sharp as Emacs.
         */
        if(list_empty(&ev->io_object.idle)) {
            goto out;
        }

        ev->io_object.obj_using = &ev->io_object.rsv;
        obj = ev->io_object.obj_using;

        obj->bufptr = obj->buf_start;

        err = bus_io->recv(ev->sockfd, \
                        (char *)obj->buf_start, \
                        BUS_BUF_SIZE, \
                        0);

        if(err <= 0) {

            bus_event->del(0, ev);

            ret = -1;
            bus_event_set(ev, BUS_IO_ERROR);

            bus_callback(ev);
            goto out;

        }

        obj->len = err;

        bus_callback(ev);

        bus_event_unset(ev, BUS_IO_READING);

    }

out:
    if(ret < 0)
        bus_close_socket(ev);

    return;

}

/*
 * bus_process_writing_sliding -
 * Process a sending io request on the bus_event_t structure from Kernel(Sliding IO).
 *
 * @ev:            The pointer of the bus_event_t structure.
 *
 */
static int bus_process_writing_sliding(bus_event_t *ev)
{
    int err;
    int ret = 0;
    bus_io_object_t *obj = bus_event_io_poll_get_queue_first(ev, &ev->io_object.using_w);

    /*
     *    看看io事件是否可写
     */
    if(bus_event_bit(ev, BUS_IO_WRITING)) {

        err = bus_io->send(ev->sockfd, \
                        (char *)obj->bufptr, \
                        obj->len, \
                        0);

        if(err <= 0) {

            bus_event->del(0, ev);

            ret = -1;
            bus_event_set(ev, BUS_IO_ERROR);

            bus_callback(ev);
            goto out;

        }

        obj->bufptr += err;
        obj->len -= err;
        if(obj->len <= 0) {

            bus_event_io_poll_remove_queue_first(ev, &ev->io_object.using_w);

            ev->io_object.obj_using = obj;

            bus_event_io_close(ev, BUS_IO_WRITING);

            bus_event_io_poll_put_object(ev, obj);

        }
        bus_event_unset(ev, BUS_IO_WRITING);
    }

out:
    if(ret < 0)
        bus_close_socket(ev);
    return ret;

}

/* End --- Sliding IO --- */


static void bus_pipe_handle(bus_event_t *ev)
{
    if(bus_event_bit(ev, BUS_IO_WRITING)) {

        if(bus_event_bit(ev, BUS_IO_WRITING)) {

            if(bus_process_sending(ev) < 0) {
                
                bus_close_socket(ev->pipe.to);

            }

        }

    }

    if(bus_event_bit(ev, BUS_IO_READING)) {

        if(bus_event_bit(ev, BUS_IO_READING)) {

            if(bus_process_recving(ev) < 0) {

                bus_close_socket(ev->pipe.to);

            }

            bus_add_sending(ev->pipe.to, bus_event_get_buf(ev), bus_event_get_size(ev), 0);

        }
    }
}

/*
 * bus_engine_loop - The looping of the engine.
 * @it_is_not_available_now:        Unused.
 *
 */
void bus_engine_loop(void *it_is_not_available_now)
{
    int err;
    int ret = 0;

    struct list_head *first;
    struct list_head *list;
    bus_event_t *ev;

    for(;;) {

        if(!bus_event) {

            bus_panic("bus_engine_loop", "burst_buffered_io.c", 5000, 0xFEEEEEEE, "No module.");

        }

        first = list = bus_event->process(0);

        do {

            ev = (bus_event_t *)container_of(list, bus_event_t, list);
            
            if(bus_event_io_status(ev, BUS_IO_SLEEPING)) {

                goto skip_next_loop;

            }
            if(bus_event_io_status(ev, BUS_IO_READING)) {

                bus_process_recving(ev);

            }
            if(bus_event_io_status(ev, BUS_IO_WRITING)) {

                bus_process_sending(ev);

            }
            if(bus_event_io_status(ev, BUS_IO_ACCEPTING)) {

                bus_process_accepting(ev);

            }
            if(bus_event_io_status(ev, BUS_IO_CONNECTING)) {

                bus_process_connecting(ev);

            }

skip_next_loop:

            list = list->next;

        } while(list != first);

    }

err_out:
    return;

}


/*
 * bus_kernel - The kernel export function of the engine.
 *
 * We will give this to the module initializer.
 */
bus_kernel_t bus_kernel = {

    /* bus_event_get_event_t */
    &bus_event_get_event, 

    /* bus_event_get_buf_t */    
    &bus_event_get_buf, 

    /* bus_event_get_size_t */
    &bus_event_get_size, 

    /* bus_event_get_data_t */
    &bus_event_get_data, 

    /* bus_event_set_data_t */
    &bus_event_set_data,

    /* bus_callback_t */
    &bus_callback, 

    /* bus_get_callback_t */
    &bus_get_callback, 

    /* bus_set_callback_t */
    &bus_set_callback, 

    /* bus_get_socket_t */
    &bus_get_socket, 

    /* bus_add_binding_t */
    &bus_add_binding, 

    /* bus_add_sending_t */
    &bus_add_sending, 

    /* bus_add_recving_t */
    &bus_add_recving, 

    /* bus_add_connecting_t */
    &bus_add_connecting, 

    /* bus_add_accepting_t */
    &bus_add_accepting, 

    /* bus_engine_loop_t */
    &bus_engine_loop

};
